package com.gis507.mapreduce.outputformat;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class AISRecordWriter extends RecordWriter<AisClassifyBean, NullWritable> {


    private FSDataOutputStream cargoShip;
    private FSDataOutputStream passengerShip;
    private FSDataOutputStream oilShip;

    public AISRecordWriter(TaskAttemptContext job) {

        //获取文件系统对象
        try {

            FileSystem fs = FileSystem.get(job.getConfiguration());
            //货船
            cargoShip = fs.create(new Path("D:\\AAUser\\dic\\Files\\testFile3\\cargoship.csv"));

            //客船
            passengerShip = fs.create(new Path("D:\\AAUser\\dic\\Files\\testFile3\\passengerShip.csv"));

            //油轮
            oilShip = fs.create(new Path("D:\\AAUser\\dic\\Files\\testFile3\\oilShip.csv"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(AisClassifyBean key, NullWritable value) throws IOException, InterruptedException {

        if ("客船".equals(key.getShipType())){

            passengerShip.writeBytes("passengerShip");
        }else if ("货船".equals(key.getShipType())){

            cargoShip.writeBytes("cargoShip");
        }else if ("油轮".equals(key.getShipType())){

            oilShip.writeBytes("oilShip");
        }else{

            oilShip.writeBytes("other");
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {

        IOUtils.closeStreams(cargoShip,passengerShip,oilShip);
    }
}
